package com.jbxie.spaceiotmqtt.mqttUtils;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * mqtt订阅消息客户端
 */
@Component
public class MyMqttClient {
    private static final Logger logger = LoggerFactory.getLogger(MyMqttClient.class);

    //@Autowired
    private static MqttClientCallback mqttClientCallback  = new MqttClientCallback();

    public static MqttClient mqttClient = null;
    private static MemoryPersistence memoryPersistence = null;
    private static MqttConnectOptions mqttConnectOptions = null;

    /**
     * 要发布的消息队列
     */
    private static LinkedBlockingQueue<String[]> queue = new LinkedBlockingQueue<String[]>();
    public static ExecutorService executorService = Executors.newFixedThreadPool(2 * Runtime.getRuntime().availableProcessors() + 1);

    /**
     * mqtt broker 连接配置,填自己的mqtt地址,及账号密码
     */
    //private static String broker; //= "tcp://192.168.31.198:1883";
    private static String clientId;
    private static String username = "xjb";
    private static String password = "123456";


    /**
     * 创建订阅客户端
     */
    public void createClient(String broker) {
        System.out.println("create mqttClient connect " + broker);
        //String clientId = UUID.randomUUID().toString().trim().replaceAll("-", "");
        clientId = "mqttServer" + String.valueOf(System.currentTimeMillis());
        System.out.println("mqtt clientId = " + clientId);
        try {
            memoryPersistence = new MemoryPersistence(); //设置持久化方式
            mqttClient = new MqttClient(broker, clientId, memoryPersistence); //MemoryPersistence设置clientid的保存形式，默认为以内存保存
            if (!mqttClient.isConnected()) {
                //logger.info("客户端连接mqtt-broker: " + broker);
                System.out.println("create mqtt clientConnection");
                try {
                    mqttConnectOptions = new MqttConnectOptions();
                    mqttConnectOptions.setCleanSession(true); //true可以安全地使用内存持久性作为客户端断开连接时清除的所有状态
                    mqttConnectOptions.setMaxInflight(100000);
                    mqttConnectOptions.setConnectionTimeout(30); //设置连接超时,单位为秒
                    //mqttConnectOptions.setKeepAliveInterval(20); //保持连接数,设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
                    if (username != null && !"".equals(username)) {
                        mqttConnectOptions.setUserName(username);
                        mqttConnectOptions.setPassword(password.toCharArray());
                    }
                    mqttClient.setCallback(mqttClientCallback); //客户端添加回调函数
                    mqttClient.connect(mqttConnectOptions);
                    System.out.println(mqttClient.isConnected());
                } catch (MqttException me) {
                    System.out.println("reason " + me.getReasonCode());
                    System.out.println("msg " + me.getMessage());
                    System.out.println("loc " + me.getLocalizedMessage());
                    System.out.println("cause " + me.getCause());
                    System.out.println("excep " + me);
                    me.printStackTrace();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 重新连接
     */
    public static void reConnect() {
        while (true) {
            try {
                if (mqttClient != null && !mqttClient.isConnected() && mqttConnectOptions != null) {
                    Thread.sleep(1000);
                    clientId = "mqttServer" + String.valueOf(System.currentTimeMillis());
                    mqttClient.connect(mqttConnectOptions);
                    logger.info("=======尝试重新连接==============");
                    break;
                }
            } catch (MqttException | InterruptedException e) {
                logger.info("=======重新连接失败:{}==============", e.toString());
                continue;
            }
        }
    }

    /**
     * 关闭连接
     */
    public void closeConnect() {
        //关闭存储方式
        if(null != memoryPersistence) {
            try {
                memoryPersistence.close();
            } catch (MqttPersistenceException e) {
                e.printStackTrace();
            }
        } else {
            System.out.println("memoryPersistence is null");
        }
        //关闭连接
        if( mqttClient != null) {
            if(mqttClient.isConnected()) {
                try {
                    mqttClient.disconnect();
                    mqttClient.close();
                } catch (MqttException me) {
                    me.printStackTrace();
                }
            } else {
                System.out.println("mqttClient is not connect");
            }
        } else {
            System.out.println("mqttClient is null");
        }
    }

    /**
     * 订阅主题
     */
    public static void subTopic(String topic, int qos) {
        if (mqttClient != null && mqttClient.isConnected()) {
            try {
                mqttConnectOptions.setWill(topic, "close".getBytes(), 2, true);
                logger.info("clientId: " + mqttClient.getClientId() + "订阅主题" + topic);
                SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
                logger.info("订阅时间========" + df.format(new Date()) + " " + System.currentTimeMillis());
                mqttClient.subscribe(topic, qos);
                mqttClient.setCallback(mqttClientCallback);
                //logger.info("订阅成功");
            } catch (MqttException me) {
                me.printStackTrace();
            }
        } else {
            System.out.println("mqttClient is Error");
        }
    }
    /**
     * 订阅主题
     */
    public static void subTopic(String[] topic, int[] qos) {
        if (mqttClient != null && mqttClient.isConnected()) {
            try {
                //logger.info("clientId: " + mqttClient.getClientId() + "订阅主题" + topic);
                System.out.println("mqttClient subscribe topic : " + topic);
                mqttClient.subscribe(topic, qos);
                mqttClient.setCallback(mqttClientCallback);
                //logger.info("订阅成功");
            } catch (MqttException me) {
                me.printStackTrace();
            }
        } else {
            System.out.println("mqttClient is Error");
        }
    }

    /**
     * 清空主题
     */
    public static void cleanTopic(String topic) {
        if (mqttClient != null && !mqttClient.isConnected()) {
            try {
                mqttClient.unsubscribe(topic);
            } catch (MqttException me) {
                me.printStackTrace();
            }
        } else {
            System.out.println("mqttClient is Error");
        }
    }

    /**
     * 发布，默认qos为0，非持久化
     * @param topic
     * @param message
     */
    public static void pubMessage(String topic, String message) {
        pubMessage(0, false, topic, message);
    }

    /**
     * 发布消息的服务质量(推荐为：2-确保消息到达一次。0-至多一次到达；1-至少一次到达，可能重复)，
     * retained 默认：false-非持久化（是指一条消息消费完，就会被删除；持久化，消费完，还会保存在服务器中，当新的订阅者出现，继续给新订阅者消费）
     * @param topic
     * @param message
     */
    public static void pubMessage(int qos, String topic, String message){
        pubMessage(qos, false, topic, message);
    }

    /**
     * 发布消息
     * @param pubTopic
     * @param message
     * @param qos
     * @param retained
     */
    public static void pubMessage(int qos, boolean retained, String pubTopic, String message) {
        if (mqttClient != null && mqttClient.isConnected()) {
            MqttMessage mqttMessage = new MqttMessage();
            mqttMessage.setQos(qos);
            mqttMessage.setRetained(retained);
            mqttMessage.setPayload(message.getBytes());

            MqttTopic topic = mqttClient.getTopic(pubTopic);

            if (topic == null) {
                logger.error("topic not exist");
            }
            MqttDeliveryToken token;
            SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
            logger.info("发送时间========" + df.format(new Date()));
            logger.info("clientId: " + mqttClient.getClientId() + "发送主题:" + topic + "成功，内容:" + message);
            try {
                token = topic.publish(mqttMessage);
                token.waitForCompletion();
                //logger.info("message is published completely!" + token.isComplete());
                System.out.println("message is published completely! "
                        + token.isComplete());
            } catch (MqttPersistenceException me) {
                logger.error("publish fail", me);
                me.printStackTrace();
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }else {
            reConnect();
        }
    }

    /**
     * 发布消息
     */
    public static void pubMessage(MqttTopic topic, MqttMessage message) throws MqttPersistenceException,
            MqttException {
        if (topic == null) {
            logger.error("topic not exist");
        }
        MqttDeliveryToken token = topic.publish(message);
        token.waitForCompletion();
        //logger.info("message is published completely!" + token.isComplete());
        System.out.println("message is published completely! "
                + token.isComplete());
    }

    /**
     * 循环从队列queue获取要发布的消息发布
     */
    public void pubMessageFromQueue() {
        //启动发布消息的线程, 循环从队列queue获取要发布的消息发布。也就是说要发消息，只要将消息写入队列queue。
        new Thread(new PublishThread()).start();
    }
    /**
     * 将要发布的消息添加到队列
     * @param topic
     * @param content
     */
    public static void publish_common(String topic, String content, int qos) {
        String[] array = {topic, content, String.valueOf(qos)};
        //要发布的消息，入队列。
        //PublishThread会循环从这个队列获取消息发布.
        queue.offer(array);
    }

    /**
     * 从队列获取要发布的信息
     */
    public static String[] poll() {
        return (String[]) queue.poll();
    }
}
